// Dart中的Stream示例
// Stream表示一系列异步事件，类似于一个异步的Iterable

import 'dart:async';

// 创建一个简单的Stream
Stream<int> countStream(int max) async* {
  for (int i = 1; i <= max; i++) {
    // 模拟异步操作
    await Future.delayed(Duration(milliseconds: 500));
    // 产生一个值
    yield i;
  }
}

// 使用Stream.fromIterable创建Stream
void streamFromIterable() {
  print('\nStream.fromIterable示例:');
  print('-------------------');

  // 从列表创建Stream
  final stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  // 监听Stream
  stream.listen(
    (data) => print('接收到数据: $data'),
    onError: (error) => print('错误: $error'),
    onDone: () => print('Stream已关闭'),
  );
}

// 使用StreamController
void usingStreamController() {
  print('\nStreamController示例:');
  print('-------------------');

  // 创建一个StreamController
  final controller = StreamController<String>();

  // 获取Stream
  final stream = controller.stream;

  // 监听Stream
  stream.listen(
    (data) => print('从controller接收: $data'),
    onDone: () => print('Controller的stream已关闭'),
  );

  // 添加数据到Stream
  controller.add('Hello');
  controller.add('Stream');
  controller.add('World');

  // 完成后关闭controller
  // 注意：在实际应用中，通常在不再需要时关闭controller
  Future.delayed(Duration(seconds: 1), () {
    controller.close();
  });
}

// 使用async/await处理Stream
Future<void> processingWithAsyncAwait() async {
  print('\n使用async/await处理Stream:');
  print('-------------------');

  final stream = countStream(5);

  // 使用await for循环处理Stream
  await for (final value in stream) {
    print('使用await for接收: $value');
  }

  print('Stream处理完成');
}

// Stream转换和操作
Future<void> transformingStreams() async {
  print('\nStream转换和操作:');
  print('-------------------');

  final stream = countStream(10);

  // 使用where过滤
  final evenStream = stream.where((x) => x % 2 == 0);

  // 使用map转换
  final squaredStream = evenStream.map((x) => x * x);

  // 使用take限制数量
  final limitedStream = squaredStream.take(3);

  // 监听最终的Stream
  await for (final value in limitedStream) {
    print('转换后的值: $value');
  }
}

// 处理Stream错误
void handlingStreamErrors() {
  print('\n处理Stream错误:');
  print('-------------------');

  // 创建一个会产生错误的Stream
  final controller = StreamController<int>();

  // 添加一些数据和错误
  controller.add(1);
  controller.add(2);
  controller.addError('发生了一个错误');
  controller.add(3);

  // 监听并处理错误
  controller.stream.listen(
    (data) => print('接收到数据: $data'),
    onError: (error) => print('错误处理: $error'),
    onDone: () => print('Stream完成'),
  );

  // 关闭controller
  Future.delayed(Duration(seconds: 1), () {
    controller.close();
  });
}

// 广播Stream
void broadcastStreams() {
  print('\n广播Stream:');
  print('-------------------');

  // 创建一个广播StreamController
  final controller = StreamController<int>.broadcast();

  // 第一个监听器
  controller.stream.listen(
    (data) => print('监听器1: $data'),
    onDone: () => print('监听器1: 完成'),
  );

  // 添加一些数据
  controller.add(1);
  controller.add(2);

  // 延迟添加第二个监听器
  Future.delayed(Duration(milliseconds: 500), () {
    controller.stream.listen(
      (data) => print('监听器2: $data'),
      onDone: () => print('监听器2: 完成'),
    );

    // 添加更多数据
    controller.add(3);
    controller.add(4);

    // 关闭controller
    Future.delayed(Duration(milliseconds: 500), () {
      controller.close();
    });
  });
}

// 合并多个Stream
Future<void> combiningStreams() async {
  print('\n合并多个Stream:');
  print('-------------------');

  // 创建三个Stream
  final stream1 = Stream.periodic(
    Duration(milliseconds: 500),
    (x) => 'S1: ${x + 1}',
  ).take(3);
  final stream2 = Stream.periodic(
    Duration(milliseconds: 700),
    (x) => 'S2: ${x + 1}',
  ).take(3);
  final stream3 = Stream.periodic(
    Duration(milliseconds: 300),
    (x) => 'S3: ${x + 1}',
  ).take(3);

  // 使用merge合并Stream
  print('使用merge:');
  final mergedStream = StreamGroup.merge([stream1, stream2, stream3]);
  await for (final value in mergedStream) {
    print('  $value');
  }
}

void main() async {
  print('Dart Stream示例');

  // 运行各种Stream示例
  streamFromIterable();
  usingStreamController();
  await processingWithAsyncAwait();
  await transformingStreams();
  handlingStreamErrors();
  broadcastStreams();
  await combiningStreams();

  // 等待所有异步操作完成
  await Future.delayed(Duration(seconds: 2));

  print('\n所有Stream示例执行完毕');
}

// StreamGroup类用于合并多个Stream
class StreamGroup<T> {
  static Stream<T> merge<T>(Iterable<Stream<T>> streams) {
    final controller = StreamController<T>();
    int count = streams.length;

    for (var stream in streams) {
      stream.listen(
        (data) => controller.add(data),
        onError: (error) => controller.addError(error),
        onDone: () {
          count--;
          if (count == 0) controller.close();
        },
      );
    }

    return controller.stream;
  }
}
